package org.apache.ocean;

import java.net.URL;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.lang.StringUtils;
import org.jdom.Element;

/**
 * Obtains transactions from a master
 * 
 * @author jasonr
 */
// TODO: check serverinfo reportable status for masterurl being master
public class ReplicationClient {
	public static Logger log = Logger.getLogger(ReplicationClient.class.getName());
	public static long WAIT_DELAY = 3 * 1000;
	public static long TASK_TIMEOUT = 2 * 60 * 1000;
	public int httpTimeout;
	private URL masterURL;
	public ReplicationInterface replicationInterface;
	public RestartableTimer replicationTimer;

	public ReplicationClient(ReplicationInterface replicationInterface) throws Exception {
		this.replicationInterface = replicationInterface;
		replicationTimer = new RestartableTimer("replicationTimer", new Task(), WAIT_DELAY, TASK_TIMEOUT);
	}

	public static interface ReplicationInterface {
		public void transactions(String shardID, List<TransactionRecord> transactionRecords);

		public TransactionID getHighestTransactionIDDatabase(String shardID);

		public void ensureShardsExist(Map<String, ShardInfo> shardInfos) throws Exception;
	}

	public void startReplicating(URL masterURL) throws Exception {
		this.masterURL = masterURL;
		if (masterURL == null)
			stopReplicating();
		//else
		//	notifyNewTransactions();
	}

	public void stopReplicating() throws Exception {
		replicationTimer.cancel(true);
	}

	public void shutdown() throws Exception {
		replicationTimer.shutdown();
	}

	// TODO: if master switch happens while creating shards they need to be
	// deleted when ensureShardsExist is called
	public class Task implements RestartableTimer.TimeoutRunnable {
		public void run(TimeoutState timeoutState) {
			if (masterURL != null) {
				try {
					HttpTimeout httpTimeout = new HttpTimeout(3 * 1000, 3 * 1000, timeoutState);
					OceanXML response = NodeClient.getShardInfosOceanXML(masterURL, httpTimeout);
					OceanXML.ServerInfo serverInfo = response.getServerInfo();
					if (!StringUtils.equals(ClusterConstants.MASTER, serverInfo.reportableStatus)) {
						log.log(Level.SEVERE, "masterURL: " + masterURL + " reportableStatus: " + serverInfo.reportableStatus);
						return;
					}
					Map<String, ShardInfo> shardInfos = new HashMap<String, ShardInfo>();
					for (Element element : XMLUtil.getChildren(response.bodyElement)) {
						ShardInfo shardInfo = new ShardInfo(element);
						TransactionID highestID = replicationInterface.getHighestTransactionIDDatabase(shardInfo.id);
						if (highestID != null) {
							if (shardInfo.highestTransactionIDDatabase.compareTo(highestID) > 0) {
								shardInfos.put(shardInfo.id, shardInfo);
							}
						} else {
							shardInfos.put(shardInfo.id, shardInfo);
						}
					}
					replicationInterface.ensureShardsExist(shardInfos);
					for (ShardInfo shardInfo : shardInfos.values()) {
						try {
							if (masterURL == null) {
								log.log(Level.SEVERE, "master url null");
								return;
							}
							if (timeoutState.notOK()) {
								log.log(Level.SEVERE, "timed out: " + timeoutState);
								return;
							}
							processShard(masterURL, shardInfo, timeoutState);
						} catch (Throwable throwable) {
							log.log(Level.SEVERE, "", throwable);
						}
					}
				} catch (Throwable throwable) {
					log.log(Level.SEVERE, "", throwable);
				}
			}
		}
    
		// TODO: multithread the transaction record writing in the shard
		private void processShard(URL url, ShardInfo shardInfo, TimeoutState timeoutState) throws Exception {
			if (shardInfo.highestTransactionIDDatabase != null) {
				HttpTimeout httpTimeout = new HttpTimeout(1000 * 2, 1000 * 60, timeoutState);
				TransactionID myHighestTransactionID = replicationInterface.getHighestTransactionIDDatabase(shardInfo.id);
				if (myHighestTransactionID == null) {
					myHighestTransactionID = new TransactionID("0");
				}
				TransactionID startID = (TransactionID) myHighestTransactionID.clone();
				startID.increment();
				CLong length = new CLong(100);
				List<TransactionRecord> transactionRecords = NodeClient.getTransactionRecords(url, shardInfo.id, startID, length, httpTimeout);
				if (transactionRecords.size() > 0) {
					log.info("transactionRecords.size: "+transactionRecords.size());
				}
				replicationInterface.transactions(shardInfo.id, transactionRecords);
			}
		}
	}

	public void notifyNewTransactions() throws Exception {
		replicationTimer.runNow();
	}
}
